package com.bw;

import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.dataproc.SplitBatchOp;
import com.alibaba.alink.operator.batch.evaluation.EvalRegressionBatchOp;
import com.alibaba.alink.operator.batch.source.CsvSourceBatchOp;
import com.alibaba.alink.operator.common.evaluation.RegressionMetrics;
import com.alibaba.alink.operator.common.evaluation.TuningRegressionMetric;
import com.alibaba.alink.params.shared.linear.LinearTrainParams;
import com.alibaba.alink.pipeline.PipelineModel;
import com.alibaba.alink.pipeline.regression.LassoRegression;
import com.alibaba.alink.pipeline.regression.LinearRegression;
import com.alibaba.alink.pipeline.regression.RidgeRegression;
import com.alibaba.alink.pipeline.tuning.GridSearchCV;
import com.alibaba.alink.pipeline.tuning.GridSearchCVModel;
import com.alibaba.alink.pipeline.tuning.ParamGrid;
import com.alibaba.alink.pipeline.tuning.RegressionTuningEvaluator;

public class Test {
    public static void main(String[] args) throws Exception {

        //（1）使用Flink批处理加载附件中的boston_housing.csv，封装数据集DataSet，提取特征features和标签label，前10条样本数据打印控制台；（3分）
        BatchOperator.setParallelism(1);
        String schemaStr="f1 double,f2 double,f3 double,f4 double,f5 double,f6 double,f7 double,f8 double,f9 double,f10 double,f11 double,f12 double,f13 double,label double";

        CsvSourceBatchOp csv=new CsvSourceBatchOp()
                .setFilePath("datafile/boston_housing.csv")
                .setSchemaStr(schemaStr)
                .setIgnoreFirstLine(true)
                .setFieldDelimiter(",");

//        csv.print(10);
//        System.out.println("csv.count() = " + csv.count());

        // （2）查看数据的详细信息,并统计数据量，自定义划分训练集和测试集（3分）
        BatchOperator<?> tranData=new SplitBatchOp().setFraction(0.8).linkFrom(csv);
        BatchOperator<?> testData=tranData.getSideOutput(0);

        // 3、自主选择使用Alink中至少两种算法构建模型，并设定初始值
        //特征列
        String [] features=new String[]{"f1","f2","f3","f4","f5","f6","f7","f8","f9","f10","f11","f12","f13"};
        //目标值
        String label="label";

        // 线性回归
        LinearRegression lr = new LinearRegression()
                .setFeatureCols(features)
                .setLabelCol(label)
                .setMaxIter(200)
                .setL1(0.3D)
                .setPredictionCol("pred")
                .enableLazyPrintModelInfo();

        BatchOperator<?> transform = lr.fit(tranData).transform(testData);
        transform.print();

        // 套索回归
        LassoRegression lasso = new LassoRegression()
                .setFeatureCols(features)
                .setLambda(0.3)
                .setMaxIter(200)
                .setLabelCol(label)
                .setPredictionCol("pred")
                .enableLazyPrintModelInfo();
        BatchOperator<?> transform1 = lasso.fit(tranData).transform(testData);
        transform1.print();

        //对以上算法参数进行调优，可以打印中间调优结果
        // 线下回归调优
        ParamGrid paramGrid = new ParamGrid()
                .addGrid(lr, LinearRegression.L_1,new Double[] {0.1,0.2})
                .addGrid(lr, LinearRegression.OPTIM_METHOD, new LinearTrainParams.OptimMethod[]{LinearTrainParams.OptimMethod.SGD, LinearTrainParams.OptimMethod.Newton});

        // 根据什么指标调优
        RegressionTuningEvaluator tuningEvaluator = new RegressionTuningEvaluator()
                .setLabelCol(label)
                .setPredictionCol("pred")
                .setTuningRegressionMetric(TuningRegressionMetric.MSE)
                .setTuningRegressionMetric(TuningRegressionMetric.RMSE);


        // 网格搜索
        GridSearchCV cv = new GridSearchCV()
                .setEstimator(lr)
                .setParamGrid(paramGrid)
                .setTuningEvaluator(tuningEvaluator)
                .setNumFolds(2)
                .enableLazyPrintTrainInfo("TrainInfo");

        // 训练模型
        GridSearchCVModel model = cv.fit(tranData);
        // 最佳模型
        PipelineModel bestPipelineModel = model.getBestPipelineModel();
        // 线性回归最好的结果
        BatchOperator<?> lr_result = bestPipelineModel.transform(testData);


        //对套索回归调优
        // 套索回归调优
        ParamGrid paramGrid1 = new ParamGrid()
                .addGrid(lasso, LassoRegression.MAX_ITER,new Integer[]{50,100})
                .addGrid(lasso, LassoRegression.OPTIM_METHOD, new LinearTrainParams.OptimMethod[]{LinearTrainParams.OptimMethod.SGD, LinearTrainParams.OptimMethod.Newton});

        // 网格搜索
        GridSearchCV cv1 = new GridSearchCV()
                .setEstimator(lasso)
                .setParamGrid(paramGrid1)
                .setTuningEvaluator(tuningEvaluator)
                .setNumFolds(2)
                .enableLazyPrintTrainInfo("TrainInfo");

        // 训练模型
        GridSearchCVModel model1 = cv1.fit(tranData);
        // 最佳模型
        PipelineModel bestPipelineModel1 = model1.getBestPipelineModel();
        BatchOperator<?> la_result = bestPipelineModel1.transform(testData);

        // 回归指标
        // 线下回归
        RegressionMetrics lr_metrics = new EvalRegressionBatchOp()
                .setPredictionCol("pred")
                .setLabelCol(label)
                .linkFrom(lr_result).collectMetrics();
        // 套索回归
        RegressionMetrics lr_metrics1 = new EvalRegressionBatchOp()
                .setPredictionCol("pred")
                .setLabelCol(label)
                .linkFrom(la_result).collectMetrics();

//        （5）自主选择Alink至少两种指标进行模型评估（6分）
        System.out.println("Total Samples Number:" + lr_metrics.getCount());
        System.out.println("SSE:" + lr_metrics.getSse());
        System.out.println("SAE:" + lr_metrics.getSae());
        System.out.println("RMSE:" + lr_metrics.getRmse());
        System.out.println("R2:" + lr_metrics.getR2());


        System.out.println("Total Samples Number:" + lr_metrics1.getCount());
        System.out.println("SSE:" + lr_metrics1.getSse());
        System.out.println("SAE:" + lr_metrics1.getSae());
        System.out.println("RMSE:" + lr_metrics1.getRmse());
        System.out.println("R2:" + lr_metrics1.getR2());

//        （6）选择较优模型对数据集预测，并打印出结果。（2分）
        if (lr_metrics.getRmse() < lr_metrics1.getRmse()){
            bestPipelineModel.transform(testData).print();
        }else{
            bestPipelineModel1.transform(testData).print();
        }

    }
}
